/*
 * bootstrap.c
 *
 *  Created on: May 20, 2012
 *      Author: burek
 */

#include "cerv/messaging.h"
#include "portutil/sockets.h"
#include "portutil/threading.h"
#include "misc/iniconfig.h"
#include "cerv/message-codes.h"
#include "cerv/node.h"
#include "cerv/location.h"
#include "cerv/logger.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define LOCALHOST_IP (16777343)

// Parametri jednog bootstrapa, nista globalno, tako da ih moze raditi vise odjednom
typedef struct
{
	unsigned int tick;
	unsigned int pingTolerance;
	unsigned int externalIP;
	unsigned int pingInterval;
	char *externalIPtxt;
	int running;
	int mastersPerLevel;
	CLocation * root;
	CUSocket * servSock;
	ININode * config;
	CUMutex * mutNodes;
	int log;
	int logAll;
} BSparams;

// Pomocni wrapper za neka void* prosljedjivanja
typedef struct
{
	BSparams *p;
	void *m;
	CMessage *msg;
	CLocation *loc;
} BSwrap;

// Modifikovani puts za wide charove
void bwputs(int id, const char * txt)
{
	char buffer[512];
	int i = 0;

	buffer[i++] = '"';
	while(*txt)
	{
		buffer[i++] = *txt;
		txt += 2;
	}
	buffer[i++] = '"';
	buffer[i++] = '\0';
	lputs(id, buffer);
}

// Logovanje odlazne poruke
void bs_logSnd(int log, const char * wmsg, int len, const char * addr, int port)
{
	char narrow[256];
	int i;
	if(len > 500) len = 500;
	for(i = 0; i < len / 2; ++i)
		narrow[i] = wmsg[i * 2];
	narrow[i] = '\0';
	if(CU_OK != cu_checkIPaddress(addr))
			lprintf(log, "Invalid address! \"%s\"", addr);
	lprintf(log, "[%15s : %5d] << \"%s\"", addr, port, narrow);
}

// Logovanje dolazne poruke
void bs_logRcv(int log, const char * wmsg, int len, const char * addr, int port)
{
	char narrow[256];
	int i;
	if(len > 500) len = 500;
	for(i = 0; i < len / 2; ++i)
		narrow[i] = wmsg[i * 2];
	narrow[i] = '\0';
	if(CU_OK != cu_checkIPaddress(addr))
		lprintf(log, "Invalid address! \"%s\"", addr);
	lprintf(log, "[%15s : %5d] >> \"%s\"", addr, port, narrow);
}

// Dodaj node u listu parametara poruke, potrebno kada bootstrap treba javiti
// novom nodu listu dosadasnjih servenata koji ga se ticu
void bs_appendNodes(CLocation * loc, void * param)
{
	BSwrap *bsw = (BSwrap*)param;
	CMessage * msg = (CMessage*)bsw->msg;
	int i;

	// Samo preci, potomci i neposredne komsije
	if(cl_isAncestorOf(loc, bsw->loc) || cl_isAncestorOf(bsw->loc, loc) ||
			cl_isSiblingOf(loc, bsw->loc))
	{
		for(i = 0; i < loc->numNodes; ++i)
		{
			if(loc->node[i]->role == 1)
			{
				char tmp[256];
				char path[256];
				cl_constructFullAddress(loc, path);
				sprintf(tmp, "%u:%d:%s", loc->node[i]->addrInt, loc->node[i]->port, path);
				cm_appendString(msg, tmp);
			}
		}
	}
}

// Registracija serventa, kada se isti upali i prvobitno javi bootstrapu
int bs_registerServent(BSparams * params, CUSocket * sock, const char * address)
{
	char tmp[256];
	char addrRev[512];
	char *tloc;
	int port;
	CLocation *loc, *parent, *cur;
	CNode *n;
	CMessage *reply;

	sscanf(address, "%d", &port);
	tloc = strchr(address, ':');
	if(tloc == NULL)
	{
		lputs(params->log, "BS) Failed to parse address");
		return 500;
	}
	tloc++;

	cl_reverseAddress(tloc, addrRev);

	// Modifikacija stabla je u pitanju
	cu_lockMutex(params->mutNodes);

	// Moze li postojati takva lokacija?
	loc = cl_createLocation(params->root, addrRev);
	if(loc == NULL)
	{
		lputs(params->log, "BS) Failed to create location");
		cu_unlockMutex(params->mutNodes);
		return 500;
	}

	if(loc->firstChild == NULL)
	{
		if(loc->numNodes > 0)
		{
			// Vec postoji servent za tu krajnju lokaciju, ne moze ih biti vise
			lprintf(params->log, "BS) Node exists : %s\n", address);
			cu_unlockMutex(params->mutNodes);
			return 502;
		}
	}
	loc->endPoint = 1;

	sprintf(tmp, "%u:%s", sock->ipAddr, address);
	n = cl_recFindNode(loc, sock->ipAddr, sock->port);

	// Ako node ne postoji vec na visim nivoima, napraviti ga
	if(n == NULL)
		n = cn_constructNode(tmp);

	// Ako je ukljucena zamjena localhost adrese, obaviti i to
	if(params->externalIP > 0 && n->addrInt == LOCALHOST_IP)
	{
		strcpy(n->addrText, params->externalIPtxt);
		n->addrInt = params->externalIP;
	}

	// Registracija se racuna kao znak zivota
	n->lastPong = params->tick;
	cur = loc;
	parent = loc->parent;

	// Proci od krajnje lokacije prema vrhu i zaduziti novi node na sva
	// slobodna mjesta za masterovanje usput
	while(parent)
	{
		if(parent->parent == NULL) break;

		if(parent->numNodes < params->mastersPerLevel)
		{
			n->role = 1;
			if(parent->parent != NULL)
			{
				cl_attachNode(parent, n, params->mastersPerLevel);
			}
		}

		cur = cur->parent;
		if(cur)
			parent = cur->parent;
		else
			parent = NULL;
	}

	cl_attachNode(loc, n, 1);

	lprintf(params->log, "BS) Node registered! IP: %s:%d (%u) @ %s", n->addrText, n->port, n->addrInt, tmp);

	// Registracija uspjela, treba odgovoriti sa spiskom ostalih nodova
	reply = cm_createMessage(CMD_REPLY_OK);

	BSwrap bsw;
	bsw.m = NULL;
	bsw.msg = reply;
	bsw.p = params;
	bsw.loc = cl_getTopLevel(n, loc);

	// Obici cijelo stablo i za svaku lokaciju pozvati bs_appendNodes
	cl_traverseTree(params->root, bs_appendNodes, &bsw);

	// Finalizovati i poslati odgovor sa sastavljenim spiskom
	cm_finalizeMessage(reply);
	cu_sendData(sock, reply->buffer, reply->length);
	if(params->logAll) bs_logSnd(params->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
	cm_destroyMessage(reply);

	// Gotova modifikacija stabla
	cu_unlockMutex(params->mutNodes);

	return 0;
}

// Servent se odjavljuje porukom
int bs_unregisterServent(BSparams * params, CUSocket * sock, const char * address)
{
	CLocation *loc, *ploc;
	char addrRev[512];
	CMessage *reply;

	cl_reverseAddress(address, addrRev);
	loc = cl_getByAddr(params->root, addrRev);

	if(loc == NULL)	// Nije ni bio registrovan?
	{
		lprintf(params->log, "BS) Address not registered in the first place - %s", address);
		return 500;
	}
	else
	{
		if(loc->numNodes == 0)	// Nikoga ni nema na toj lokaciji?
		{
			lprintf(params->log, "BS) Address currently not occupied, cannot unregister - %s", address);
			return 500;
		}
		else
		{
			CNode *node = NULL;
			cu_lockMutex(params->mutNodes);
			if(loc->firstChild == NULL)
				node = loc->node[0];

			if(node == NULL)	// Potrazi node, mozda ga vec nema
			{
				lprintf(params->log, "BS) Node not registered, cannot unregister - %s", address);
				cu_unlockMutex(params->mutNodes);
				return 500;
			}
			lprintf(params->log, "BS) Unregistered node - %s", address);
			cl_detachNode(loc, node);
			ploc = loc->parent;

			// Ukloni ga duz cijelog stabla
			while(ploc)
			{
				if(cl_detachNode(ploc, node))
					ploc = ploc->parent;
				else
					break;
			}
			cn_destroyNode(node);

			// Odgovori potvrdno, odjava je uspjela
			reply = cm_createMessage(CMD_REPLY_OK);
			cm_finalizeMessage(reply);
			cu_sendData(sock, reply->buffer, reply->length);
			if(params->logAll) bs_logSnd(params->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
			cm_destroyMessage(reply);

			cu_unlockMutex(params->mutNodes);
			return 0;
		}
	}
}

// Posalji pingove nodovima na datoj lokaciji i skupi pongove
void bs_sendPings(BSparams * params, CLocation * root)
{
	CUSocket * sock = NULL;
	CLocation * child = NULL;
	CMessage *msg = NULL, *reply = NULL;
	char recv[128];
	char num[16];
	int reclen;

	// Broj koji ce se slati je 1000000 + broj sekundi, tako da svaki put bude jedinstveno
	sprintf(num, "%u", 1000000 + params->tick);
	if(root == NULL) return;

	// Pripremi poruku, svima ce biti ista
	msg = cm_createMessage(CMD_PING);
	cm_appendString(msg, num);
	cm_finalizeMessage(msg);

	if(root->numNodes > 0)
	{
		int i;
		for(i = 0; i < root->numNodes; ++i)
		{
			if(root->node[i]->lastPong < params->tick)
			{
				// Otvaranje socketa
				sock = cu_createTCPSocket();

				if(cu_openClientSocket(sock, root->node[i]->addrText, root->node[i]->port) == CU_OK)
				{
					cu_sendData(sock, msg->buffer, msg->length);
					if(params->logAll) bs_logSnd(params->log, msg->buffer, msg->length, sock->ipTxt, sock->port);
					reclen = cu_recvSpin(sock, recv, 128, 20, 20);
					if(params->logAll) bs_logRcv(params->log, recv, reclen, sock->ipTxt, sock->port);
					reply = cm_decodeMessage(recv, reclen);

					// Ako je odgovor stigao, uporedi da li je isti broj vracen
					if(reply != NULL && reply->code == CMD_PONG)
					{
						if(strcmp(reply->pVal[0].sval, num) == 0)
						{
							root->node[i]->lastPong = params->tick;
						}
						else
						{
							lprintf(params->log, "BS) Got wrong pong (%s) vs (%s) from %s : %d", num, reply->pVal[0].sval, root->node[i]->addrText, root->node[i]->port);
						}
					}
					cm_destroyMessage(reply);
				}
				else
				{
					lprintf(params->log, "BS) FAILED ping to %s : %d\n", root->node[i]->addrText, root->node[i]->port);
				}
				cu_closeSocket(sock);
				free(sock);
			}
		}
	}

	child = root->firstChild;
	while(child)
	{
		bs_sendPings(params, child);
		child = child->next;
	}

	cm_destroyMessage(msg);
}

// Provjeri ima li neko da se uspavao
void bs_checkPing(CLocation * root, void * vparams)
{
	BSparams * params = (BSparams*)vparams;
	int i;

	cu_lockMutex(params->mutNodes);
	for(i = 0; i < root->numNodes; ++i)
	{
		// Ako predugo nije bilo ponga, dropuj ga
		if(root->node[i]->lastPong - params->tick > params->pingTolerance)
		{
			CNode *nodeToDrop = root->node[i];
			lprintf(params->log, "BS) Dropping node [%d] : %s:%d", i, root->node[i]->addrText, root->node[i]->port);
			cl_detachNode(root, root->node[i]);
			//cn_destroyNode(nodeToDrop);
		}
	}
	cu_unlockMutex(params->mutNodes);
}

void bs_checkPings(BSparams * params, CLocation * root)
{
	cl_traverseTree(root, bs_checkPing, params);
}

// Stigla poruka - sta s njom?
int bs_processMessage(CMessage * msg, CUSocket * sock, BSparams * params)
{
	int result = 0;
	if(msg == NULL) return 1;

	switch(msg->code)
	{
	case CMD_BOOTSTRAP_REGISTER_SERVENT:
		result = bs_registerServent(params, sock, msg->pVal[0].sval);
		break;

	case CMD_BOOTSTRAP_UNREGISTER_SERVENT:
		result = bs_unregisterServent(params, sock, msg->pVal[0].sval);
		break;

	case CMD_POKE:
		lprintf(params->log, "BS) Got poked by %s:%d", sock->ipTxt, sock->port);
		break;

	default:
		lprintf(params->log, "Unknown message code %d", msg->code);
		break;
	}

	return result;
}

// Za "nestandardne", tekstualne poruke, rucno utipkane u terminal
void bs_altHandler(BSparams * bp, const char * buffer, int length)
{
	char text[1024];
	int i;
	char *param;

	for(i = 0; i < length / 2; ++i)
		text[i] = buffer[i * 2];
	text[length / 2] = '\0';
	length /= 2;

	param = strchr(text, ' ');
	if(param)
	{
		*param = '\0';
		param++;
	}

	if(strcmp("dumpTree", text) == 0)	// Listanje stabla
	{
		cl_dumpTree(bp->log, bp->root);
	}
	else if(strcmp("logTraffic", text) == 0)	// Da li da ispisuje sve poruke?
	{
		if(param != NULL)
			bp->logAll = atoi(param);
		else
			lprintf(bp->log, "logTraffic = %d", bp->logAll);
	}
	else if(strcmp("exit", text) == 0)	// Gasi
	{
		bp->running = 0;
	}
	else if(strcmp("pingInterval", text) == 0)	// Interval pingovanja servenata u sekundama
	{
		if(param != NULL)
			bp->pingInterval = atoi(param);
		else
			lprintf(bp->log, "pingInterval = %d sec(s)", bp->pingInterval);
	}
	else if(strcmp("pingTolerance", text) == 0)	// Koliko pongovi smiju kasniti?
	{
		if(param != NULL)
			bp->pingTolerance = atoi(param);
		else
			lprintf(bp->log, "pingTolerance = %d sec(s)", bp->pingTolerance);
	}
	else if(strcmp("relay", text) == 0)	// Prosljedjivanje poruke
	{
		if(param != NULL)
		{
			char *port = strchr(param, ':');
			if(port != NULL)
			{
				*port = '\0';
				port++;
				if(CU_OK == cu_checkIPaddress(param))	// Provjeri adresu i port
				{
					char *msgBegin = strchr(port, ' ');
					if(msgBegin != NULL)
					{
						int rport = 0;
						*msgBegin = '\0';
						msgBegin++;

						rport = atoi(port);
						if(rport > 0)
						{
							// Otvori socket i salji
							CUSocket *sock = cu_createTCPSocket();
							if(CU_OK == cu_openClientSocket(sock, param, rport))
							{
								char wideBuff[512];
								int len = strlen(msgBegin);
								int i;
								if(len > 511) len = 511;

								// Napravi wide charove
								for(i = 0; i < len; ++i)
								{
									wideBuff[i * 2] = msgBegin[i];
									wideBuff[i * 2 + 1] = '\0';
								}

								cu_sendData(sock, wideBuff, len * 2);
								lprintf(bp->log, "Relay successful to %s:%d", param, rport);

								// Sacekaj malo na odgovor
								len = cu_recvSpin(sock, wideBuff, 512, 20, 20);
								if(len > 0)
								{
									lprintf(bp->log, "Relay - got response:");
									bwputs(bp->log, wideBuff);
								}
								else
								{
									lprintf(bp->log, "Relay - no response");
								}
							}
							else
							{
								lprintf(bp->log, "Relay - error connectiong to %s:%d", param, rport);
							}
							cu_closeSocket(sock);
							free(sock);
						}
						else
						{
							lprintf(bp->log, "Relay - invalid port \"%s\"", port);
						}
					}
					else
					{
						lprintf(bp->log, "Relay - no message specified");
					}
				}
				else
				{
					lprintf(bp->log, "Relay - invalid IP address: \"%s\"", param);
				}
			}
			else
			{
				lprintf(bp->log, "Relay - invalid IP address: \"%s\"", param);
			}
		}
		else
		{
			lprintf(bp->log, "Relay - no parameters specified, format: relay address:port message");
		}
	}
	else
	{
		lputs(bp->log, "Unrecognized non-standard message");
	}
}

// Opsluzivanje zahtjeva
void * bs_requestHandler(void * params)
{
	BSwrap *w = (BSwrap*)params;
	BSparams *p = w->p;
	CUSocket *sock = (CUSocket*)w->m;
	CMessage *msg, *reply;
	char buffer[4096];
	int len = 0;
	int result = 1;

	// Primi poruku
	len = cu_recvData(sock, buffer, 4096);
	if(p->logAll) bs_logRcv(p->log, buffer, len, sock->ipTxt, sock->port);

	// Ako je tekstualna, probaj altHandlerom
	if(len > 0)
	{
		if(buffer[0] < '0' || buffer[0] > '9')
		{
			bs_altHandler(p, buffer, len);
			return NULL;
		}
	}

	// Dekoduj (navodno) standardnu poruku
	msg = cm_decodeMessage(buffer, len);

	if(msg == NULL)
	{
		lprintf(p->log, "BS) Malformed message received, ignoring");
		cu_closeSocket(sock);
		free(sock);
	}
	else
	{
		// Ako se moze procitati, procesiraj
		result = bs_processMessage(msg, sock, p);

		// Ako postoji greska, javi
		if(result > 0)
		{
			reply = cm_createMessage(CMD_REPLY_FAIL);
			cm_appendInt(reply, result);
			cm_finalizeMessage(reply);
			cu_sendData(sock, reply->buffer, reply->length);
			if(p->logAll) bs_logSnd(p->log, reply->buffer, reply->length, sock->ipTxt, sock->port);
			cm_destroyMessage(reply);
			lprintf(p->log, "BS) Reply sent - FAIL");
		}

		cu_closeSocket(sock);
		free(sock);
	}

	cm_destroyMessage(msg);
	return NULL;
}

// Beskonacno osluskivanje dolaznih konekcija, kada nesto dodje,
// spawnuj thread za servisiranje
void * bs_socketListener(void * params)
{
	BSparams *p = (BSparams*)params;
	CUSocket *sock = p->servSock, *client = NULL;

	while(p->running)
	{
		BSwrap *w = (BSwrap*) malloc(sizeof(BSwrap));
		client = cu_acceptConnection(sock);

		w->p = p;
		w->m = client;
		cu_spawnThread(bs_requestHandler, (void*)w);
	}

	return NULL;
}

// Entry point za bootstrap
void * bootstrapInstanceEntry(void * params)
{
	CUSocket *svSock = cu_createTCPSocket();
	CUThread *listenerThread = NULL;
	BSparams bsParams;
	ININode *config = ini_parseFile("cervent.ini");	// Procitaj konfiguraciju

	int bsPort = ini_getInt(config, "bootstrap/port", 9666);

	bsParams.log = l_register();	// Registruj se na logger i postavi naslov
	l_setTitle(bsParams.log, "[BOOT]");
	l_registerPort(bsParams.log, bsPort);

	// Ne moze raditi bez socketa
	if(!svSock)
		{ lprintf(bsParams.log, "Cannot start server"); l_disconnect(bsParams.log); return NULL; }
	if(cu_openServerSocket(svSock, bsPort) == CU_ERROR)
		{ lprintf(bsParams.log, "Cannot start server"); l_disconnect(bsParams.log); return NULL; }

	// Inicijalizuj sva polja
	bsParams.root = cl_createRoot();
	bsParams.running = 1;
	bsParams.config = config;
	bsParams.servSock = svSock;
	bsParams.tick = 0;
	bsParams.pingTolerance = 20;
	bsParams.mastersPerLevel = 3;
	bsParams.mutNodes = cu_createMutex();
	bsParams.externalIP = 0;
	bsParams.externalIPtxt = NULL;
	bsParams.logAll = ini_getInt(config, "logging/log_traffic", 0);
	bsParams.pingInterval = (unsigned int)ini_getInt(config, "bootstrap/ping_interval", 5);

	// Da li je postavljena zamjena localhost adrese?
	if(ini_getString(config, "bootstrap/externalIP", NULL) != NULL)
	{
		bsParams.externalIPtxt = (char*)calloc(1, 16);
		strcpy(bsParams.externalIPtxt, ini_getString(config, "bootstrap/externalIP", NULL));
	}


	// Procitaj zamjensku adresu
	if(bsParams.externalIPtxt != NULL)
	{
		int p1, p2, p3, p4;
		if(4 == sscanf(bsParams.externalIPtxt, "%d.%d.%d.%d", &p1, &p2, &p3, &p4))
		{
			bsParams.externalIP = (p1 & 0xFF) + ((p2 & 0xFF) << 8) + ((p3 & 0xFF) << 16) + ((p4 & 0xFF) << 24);
			lprintf(bsParams.log, "Text IP : \"%s\", NumIP : %u", bsParams.externalIPtxt, bsParams.externalIP);
		}
		else
		{
			lprintf(bsParams.log, "Invalid IP in config: \"%s\"", bsParams.externalIPtxt);
			free(bsParams.externalIPtxt);
			bsParams.externalIPtxt = NULL;
			bsParams.externalIP = 0;
		}
	}

	// Pocni opsluzivati zahtjeve
	listenerThread = cu_spawnThread(bs_socketListener, &bsParams);

	lprintf(bsParams.log, "Bootstrap started at port %d", bsPort);

	while(bsParams.running)
	{
		bsParams.tick++;	// Bootstrap otkucava svake sekunde
		cu_sleep(1000);

		// Ako je vrijeme za pingovanje, pinguj i provjeri rezultate
		if(bsParams.pingInterval > 0 && bsParams.tick % bsParams.pingInterval == 0)
		{
			bs_sendPings(&bsParams, bsParams.root);
			bs_checkPings(&bsParams, bsParams.root);
		}
	}

	// Ubij listenera, kraj rada
	cu_killThread(listenerThread);

	ini_destroy(config);
	exit(0);
	return NULL;
}
